/*
 * ConexionPD.c
 *
 *  Created on: 26/10/2011
 *      Author: utn_so
 */

#include "libPRAID.h"

void setearFD (struct pollfd *descriptores,int32_t nrosocketFS)
{
	int32_t i=0;
	while(descriptores[i].fd != 0 )
	{
		i++;
	}
	descriptores[i].fd = nrosocketFS;
	descriptores[i].events = POLLIN;
	descriptores[i].revents = 0;
}


void distribuirPedidos ()
{
	extern sem_t semaforoLista;
	int32_t socketFS;
	int32_t estado;
	t_NIPC paquete;

	while (1)
	{
		sem_wait(&semaforoLista);
		desencolarPeticionColaGral(&socketFS, &estado, &paquete);

		if (paquete.PayloadDescriptor == ESCRIBIRSECTOR)
		{
			if (socketFS == -1)
			{
				buscarNodosSinSincroYEncola(socketFS, estado, paquete);
				NIPC_destruirpaquete(&paquete);
			}
			else
			{
				encolarEnTodosLosDiscos (socketFS, estado, paquete);
				NIPC_destruirpaquete(&paquete);
			}
		}

		else if (paquete.PayloadDescriptor == LEERSECTOR)
		{
			encolarPedidoLectura(socketFS, estado, paquete);
			NIPC_destruirpaquete(&paquete);
		}
	}
}

tNodoLista *agregarDisco (int32_t nroSocketPD, int32_t *idDisco)
{
	int32_t valorRetorno;
	extern tLista *listaDiscos;
	tNodoLista *nodoAux = NULL;
	extern int32_t cantDiscos;
	extern int32_t sincro;
	pthread_mutex_t bloqueaColaDisco;
	pthread_mutex_t bloqueaCantPedidos;
	sem_t semaforoDisco;
	valorRetorno = sem_init(&semaforoDisco, 0, 0);
	pthread_mutex_init (&bloqueaColaDisco, NULL);
	pthread_mutex_init (&bloqueaCantPedidos, NULL);

	tNodoLista *nuevoNodo = NULL;
	nuevoNodo = malloc(sizeof(tNodoLista));
	nuevoNodo->idDisco = *idDisco;
	nuevoNodo->nroSocketPD = nroSocketPD;
	nuevoNodo->colaDisco = crearColaPeticiones ();
	nuevoNodo->cantPedidos = 0;
	nuevoNodo->semaforos.bloqueaColaDisco = bloqueaColaDisco;
	nuevoNodo->semaforos.bloqueaCantPedidos = bloqueaCantPedidos;
	nuevoNodo->semaforos.semaforoDisco = semaforoDisco;
	nuevoNodo->siguiente = NULL;
	discosConectados();
	if (cantDiscos == 0)
		nuevoNodo->sincro = sincro;
	else
		nuevoNodo->sincro = 0;

	/* Si la lista está vacía */
	   if(listaDiscos->head == NULL )
	   {
	      /* Añadimos la lista a continuación del nuevo nodo */
		   nuevoNodo->siguiente = listaDiscos->head;
	      /* Ahora, el comienzo de nuestra lista es en nuevo nodo */
		   listaDiscos->head = nuevoNodo;
	   }
	   else
	   {
		   nodoAux = listaDiscos->head;
	      /* Avanzamos hasta el último elemento  */
	      while(nodoAux->siguiente)
	    	  nodoAux = nodoAux->siguiente;
	      /* Insertamos el nuevo nodo después del nodo anterior */
	      nuevoNodo->siguiente = nodoAux->siguiente;
	      nodoAux->siguiente = nuevoNodo;
	   }

	return nuevoNodo;
}

void iniciarComunicacionPD (void *parametro)
{
	tNodoLista *disco = (tNodoLista *) parametro;
	uint32_t cantSincronizada = 0;
	int32_t retorno = 0;
	int32_t contador = 0;
	int32_t *bitVector;
	int32_t tamBitVector = 64; // modificar tamaño
	int32_t cantParcialSincronizada = tamBitVector;
	char pTiempo [18];
	time_t tiempo;
	struct tm *tlocal;
	extern int32_t cantDiscos;
	extern sem_t semaforoLista;
	extern tConfig config;
	extern t_log* loguear;
	extern int32_t sincro;
	int32_t valor;
	int32_t bool = 0;
	bitVector = (int32_t *)calloc(tamBitVector,sizeof(int32_t));
	t_NIPC paquete;

	while (retorno != -1)
	{
		discosConectados ();
		int32_t j=0;
		if (contador == 0 && config.flagConsola == 0 && (cantDiscos > 1 ) && disco->sincro < sincro)
		{
		        tiempo = time (NULL);
		        tlocal = localtime (&tiempo);
			strftime (pTiempo, 18, "%d/%m/%y %H:%M:%S", tlocal);
			log_info(loguear, "PRAID", "Inició la sincronización del disco %d. Tiempo de inicio: %s", disco->idDisco, pTiempo);
		}

		valor = bitVectorCompleto(bitVector, tamBitVector);
		if ((valor == 1) && (disco->sincro < sincro) && (cantDiscos > 1 ))
		{
			disco->sincro = disco->sincro + tamBitVector;
			if (disco->sincro == sincro && config.flagConsola == 0 && (cantDiscos > 1 )) //flagConsola == 1
			{
			    tiempo = time (NULL);
			    tlocal = localtime (&tiempo);
				strftime (pTiempo, 18, "%d/%m/%y %H:%M:%S", tlocal);
				log_info(loguear, "PRAID", "Se terminó la sincronización del disco %d. Tiempo de finalización: %s", disco->idDisco, pTiempo);

			}
			while (j < tamBitVector)
			{
				bitVector[j] = 0;
				j++;
			}
			bool = 0;
			valor = 0;
		}
		if ((valor == 0) && (disco->sincro < sincro) && (cantDiscos > 1 ) && (bool == 0))
		{
			cantParcialSincronizada = 0;
			int32_t i=0;
			bool = 1;
			while (i < tamBitVector)
			{
				NIPC_crearpaquete (NULL, LEERSECTOR, 0, NULL, cantSincronizada, &paquete);
				encolarPeticionColaGral (-1, ESPERA, paquete);
				NIPC_destruirpaquete(&paquete);
				sem_post(&semaforoLista);
				cantSincronizada++;
				i++;
			}
			contador++;
		}

		sem_wait (&disco->semaforos.semaforoDisco);
		retorno = enviarPeticionesADisco (disco->nroSocketPD, disco->colaDisco, disco->idDisco, disco);
		if (retorno != -1 )
			retorno = recibirRespuestasPD (disco->nroSocketPD, disco, &cantParcialSincronizada, &bitVector, contador, tamBitVector);
	}
}


void recibirPedidosFS(int32_t descriptorFS,t_NIPC paquete)
{
	extern sem_t semaforoLista;
	encolarPeticionColaGral (descriptorFS, ESPERA, paquete);
	sem_post(&semaforoLista);
}

int32_t enviarPeticionesADisco (int32_t nroSocketPD, tCola *colaDePeticiones, int32_t idDisco, tNodoLista *disco)
{
	extern tConfig config;
	extern t_log* loguear;
	tNodo *ptrNodoCola = NULL;
	ptrNodoCola = colaDePeticiones->ant;
	int32_t retorno = 0;
	int32_t dataRecibido;
	int32_t bloquearCola;

	bloquearCola = pthread_mutex_lock(&(disco->semaforos.bloqueaColaDisco));

	while (ptrNodoCola != NULL)
	{
		if (ptrNodoCola->estado == ENVIADO)
			ptrNodoCola = ptrNodoCola->siguiente;
		else if (ptrNodoCola->estado == ESPERA)
		{
			dataRecibido = NIPC_enviar (nroSocketPD, ptrNodoCola->paquete);
			if (config.flagConsola == 1)
				if (ptrNodoCola->paquete.PayloadDescriptor == LEERSECTOR)
					log_info(loguear, "PRAID", "Se hizo un read del disco %d", idDisco);

			if (dataRecibido == 0) // para ver si se cayo un disco
			{
				seCayoUnDisco (nroSocketPD, colaDePeticiones);
				retorno = -1;
				return retorno;
			}
			else
				ptrNodoCola->estado = ENVIADO;
			ptrNodoCola = ptrNodoCola->siguiente;
		}
	}

	bloquearCola = pthread_mutex_unlock(&(disco->semaforos.bloqueaColaDisco));
	return retorno;
}

int32_t recibirRespuestasPD (int32_t socketConexionPD, tNodoLista *ptrDisco, int32_t *cantParcialSincronizada, int32_t **bitVector, int32_t contador, int32_t tamBitVector)
{
	extern sem_t semaforoLista;
	t_NIPC paquete;
	int32_t tamPaquete;
	int32_t valorDeRetorno;
	int32_t retorno = 0;

	tamPaquete = NIPC_recibir (socketConexionPD, &paquete);

	if (tamPaquete == 0) // para ver si se cayo un disco
	{
		seCayoUnDisco (socketConexionPD, ptrDisco->colaDisco);
		retorno = -1;
		return retorno;
	}
	else
	{
		int32_t socketFS=0, estado=0;
		buscarPaqueteYDesencolar (paquete, ptrDisco, &socketFS, &estado);
		if (socketFS == -1 && paquete.PayloadDescriptor == LEERSECTOR)
		{
			paquete.PayloadDescriptor = ESCRIBIRSECTOR;
			encolarPeticionColaGral(socketFS, ESPERA, paquete);
			NIPC_destruirpaquete(&paquete);
			sem_post (&semaforoLista);
		}
		else if ((socketFS == -1) && (paquete.PayloadDescriptor == ESCRIBIRSECTOR) && (paquete.nroSector >= (contador-1)*tamBitVector) && (paquete.nroSector < contador*tamBitVector)) //paquete.nroSector> a lo q tengo sincro
		{
			actualizarBitVector (bitVector, paquete, tamBitVector);
			(*cantParcialSincronizada) ++;
			NIPC_destruirpaquete(&paquete);
		}
		else if (socketFS != -1)
		{
			valorDeRetorno = NIPC_enviar (socketFS, paquete);
			NIPC_destruirpaquete (&paquete);
		}
	}

	return retorno;
}


void seCayoUnDisco (int32_t nroSocketPD, tCola *colaDePeticiones)
{
	extern int32_t cantDiscos;
	extern t_log *loguear;
	extern tConfig config;
	int32_t valorRetorno;
	desencolarDisco(nroSocketPD);
	discosConectados ();
	valorRetorno = algunDiscoSincronizado();
	if (cantDiscos > 0 && valorRetorno == 1)
	{
		redistribuirPedidosDiscoCaido (colaDePeticiones); // los encola en colaGral
	}
	else
	{
		if (config.flagConsola == 1)
			log_info(loguear, "PRAID", "No hay ningún disco disponible, el sistema finalizará su ejecución");
		exit(0);
	}
}

#define HANDSHAKE 0

int32_t handshakePPD (int32_t socketPD, int32_t *idDisco)
{
	extern tLista *listaDiscos;
    extern t_log *loguear;
    extern tConfig config;
    extern int32_t cantDiscos;
    extern tDiscoCHS chs;
    t_NIPC paquete, paqueteRecibido, paqueteNuevo;
    int32_t valorRetorno = -1;
    int32_t retorno=0, cantidad;
    int32_t *informacion;
    tNodoLista *ptrNodoLista = NULL;
    ptrNodoLista = listaDiscos->head;
    tDiscoCHS chsRecibido;
    NIPC_crearpaquete (NULL, HANDSHAKE, 0, NULL, 0,&paquete);

    valorRetorno = NIPC_recibir(socketPD, &paqueteRecibido);

    informacion = (int32_t *)calloc(4,sizeof(int32_t));
    informacion = parsearSectores(paqueteRecibido.payload, &cantidad);
  	*idDisco = informacion[0];
  	chsRecibido.c = informacion[1];
  	chsRecibido.h = informacion[2];
  	chsRecibido.s = informacion[3];

  	discosConectados();
  	if (cantDiscos == 0)
  	{
  		chs.c = informacion[1];
  		chs.h = informacion[2];
  		chs.s = informacion[3];
  	}

    if ( chsRecibido.c == chs.c && chsRecibido.h == chs.h && chsRecibido.s == chs.s)
    {
    	int32_t bandera = 0;
    	while (ptrNodoLista != NULL && bandera != 1)
    	{
    		if (ptrNodoLista->idDisco == *idDisco)
    			bandera = 1;
    		else
    			ptrNodoLista = ptrNodoLista->siguiente;
    	}
    	if (bandera==1)
    	{
    		char mensaje[] = "La negociación fue fallida dado que el id no es válido";
    		NIPC_crearpaquete(NULL, ERROR, strlen(mensaje) + 1, mensaje, 0,&paqueteNuevo);
    		valorRetorno = NIPC_enviar(socketPD, paqueteNuevo);
    		close(socketPD);
    	}
    	else if (valorRetorno > 0)
    	{
    		if (config.flagConsola == 1)
    			log_info(loguear, "PRAID", "Se conectó un nuevo Planificador de Disco, su identificador es %d", *idDisco);
    		char mensaje[] = "Se conectó el nuevo Planificador de Disco";
    		NIPC_crearpaquete(NULL, ERROR, strlen(mensaje) + 1, mensaje, 0,&paqueteNuevo);
    		valorRetorno = NIPC_enviar(socketPD, paqueteNuevo);
    		NIPC_destruirpaquete(&paqueteNuevo);
    		retorno = 1;
    	}
    else
	{
		char mensaje[] = "La negociación fue fallida dado que el chs no es válido";
		NIPC_crearpaquete(NULL, ERROR, strlen(mensaje) + 1, mensaje, 0,&paqueteNuevo);
		valorRetorno = NIPC_enviar(socketPD, paqueteNuevo);
		NIPC_destruirpaquete(&paqueteNuevo);
		close(socketPD);
	}
    }
    return (retorno);
}

int32_t handshakePFS (int32_t socketFS, int32_t *idDisco)
{
    extern t_log *loguear;
    extern tConfig config;
    t_NIPC paquete, paqueteRecibido;
    int32_t valorRetorno = 0;
    t_NIPC paqueteNuevo;
    NIPC_crearpaquete (NULL, HANDSHAKE, 0, NULL, 0,&paquete);
    discosConectados();
    extern int32_t cantDiscos;
    valorRetorno = NIPC_recibir(socketFS, &paqueteRecibido);
    if (valorRetorno > 0 && cantDiscos>0 && paqueteRecibido.PayloadLength == 0)
    	valorRetorno = NIPC_enviar(socketFS, paquete);
    else if (cantDiscos<1)
	{
		char mensaje[] = "La negociación fue fallida dado que el Proceso RAID no está activo";
		NIPC_crearpaquete(NULL, HANDSHAKE, strlen(mensaje) + 1, mensaje, 0,&paqueteNuevo);
		valorRetorno = NIPC_enviar(socketFS, paqueteNuevo);
		valorRetorno = -1;
		close(socketFS);
	}
    else
    {
    	char mensaje[] = "La negociación fue fallida, la conexión se cerrará";
    	NIPC_crearpaquete(NULL, HANDSHAKE, strlen(mensaje) +1, mensaje, 0,&paqueteNuevo);
    	valorRetorno = NIPC_enviar(socketFS, paqueteNuevo);
    	valorRetorno = -1;
    	close(socketFS);
    }

    NIPC_destruirpaquete(&paquete);
    return valorRetorno;
}
